package zju.dgm.service.election;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.leader.LeaderSelector;
import org.apache.curator.framework.recipes.leader.LeaderSelectorListener;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.utils.ZKPaths;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import zju.dgm.dto.DataFrame;
import zju.dgm.service.ServiceEntry;
import zju.dgm.service.name.NameService;
import zju.dgm.util.CacheManager;
import zju.dgm.util.Paths;
import zju.dgm.util.SyncUtil;


import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Deque;
import java.util.Objects;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Future;


@Component
public class ElectionService extends ServiceEntry implements MasterUtilProvider {

    private static final Logger logger = LoggerFactory.getLogger(ElectionService.class);

    private CountDownLatch updateLatch = new CountDownLatch(1);


    @Resource
    Paths paths;

    @Autowired
    MasterThread masterThread;


    @Resource
    CuratorFramework client;

    @Autowired
    CacheManager cacheManager;

    Deque<Runnable> masterTaskList = new ConcurrentLinkedDeque<>();


    public static CountDownLatch finLatch = new CountDownLatch(1);

    @Override
    public String getRootPath() {
        return paths.getElectionPath();
    }

    @Override
    protected CountDownLatch[] getPrerequisites() {
        return new CountDownLatch[]{
                NameService.finLatch
        };
    }


    @Override
    protected void registerListener() {
        try {
            ZKPaths.mkdirs(client.getZookeeperClient().getZooKeeper(), paths.getMasterInfoPath());

            byte[] masterInfo = client.getData().forPath(paths.getMasterInfoPath());
//            logger.info("serverInfoMap中的信息：{},当前的masterInfo：{}",SyncUtil.serverInfoMap,new String(masterInfo));
            if (masterInfo != null && masterInfo.length != 0) {
                // 上个启动的残留或是已有master
                if (!SyncUtil.serverInfoMap.containsKey(new String(masterInfo))) {
                    client.setData().forPath(paths.getMasterInfoPath(), null);
                    logger.info("检查到上个启动的残留,已清除");
                } else {
                    logger.info(new String(masterInfo));
                    logger.info("检查到master已存在，选举服务提前结束");
                    SyncUtil.masterName = new String(masterInfo);
                    logger.info("对比,当前:{},master:{}", SyncUtil.serverName, SyncUtil.masterName);
                    finLatch.countDown();
                }
            }

        } catch (Exception e) {
            e.printStackTrace();
        }
        // 读取初始值
        NodeCache masterCache = cacheManager.getNodeCache(paths.getMasterInfoPath(), false);
        masterCache.getListenable().addListener(() -> {
            logger.info("检测到master信息节点变化");
            if (masterCache.getCurrentData() != null
                    && masterCache.getCurrentData().getData() != null
                    && masterCache.getCurrentData().getData().length != 0) {
                SyncUtil.masterName = new String(client.getData().forPath(paths.getMasterInfoPath()));
                logger.info("现在master是" + SyncUtil.masterName + "了");
                updateLatch.countDown();
                /*
                两种情况countDown：
                1. 监听器检查到master被写入
                2. master节点写入

                 */
                finLatch.countDown();
            } else {
                logger.info("空数据的初始化");
            }

        });
        logger.info("选举服务监听器注册完成");
    }

    @Override
    public void start() {


        try {
            ZKPaths.mkdirs(client.getZookeeperClient().getZooKeeper(), paths.getMasterInfoPath());
        } catch (Exception e) {
            e.printStackTrace();
        }
        LeaderSelector selector = new LeaderSelector(client,
                paths.getElectionPath(),
                new LeaderSelectorListener() {
                    @Override
                    public void takeLeadership(CuratorFramework curatorFramework) throws Exception {
                        logger.info(SyncUtil.serverName + " 成为leader");
                        SyncUtil.masterName = SyncUtil.serverName;
                        client.setData()
                                .forPath(paths.getMasterInfoPath(),
                                        SyncUtil.serverName.getBytes(StandardCharsets.UTF_8));
                        updateLatch.await();
                        masterThread.start();
                        updateLatch= new CountDownLatch(1);
                        masterThread.join();
                        client.setData().forPath(paths.getMasterInfoPath(), null);
                        System.out.println("退出leader");
                    }

                    @Override
                    public void stateChanged(CuratorFramework client, ConnectionState newState) {
                        System.out.println(newState + ", 变成这个了");
                    }
                });
        selector.start();
        selector.autoRequeue();// 加上这个确保原来是master的client还能抢master
        logger.info("选举服务初始化完成");
        // 确保会countDown
        try {
            Thread.sleep(10 * 1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            finLatch.countDown();
        }


    }


    @Override
    public boolean masterAvailable() {
        return Objects.equals(SyncUtil.masterName, SyncUtil.serverName);
    }

    @Override
    public void addMasterTask(Runnable task) {
        if (masterThread.isAlive()) {
            logger.info("当前server是maseter，直接提交执行任务");
            masterThread.addTask(task);
        } else {
            logger.info("当前server不是master，暂存任务");
            masterTaskList.push(task);
        }
    }

}
